package com.yesep.learn.mqtt.client.common.event;

import io.netty.handler.codec.mqtt.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 事件通知组件
 * @Description:
 */
public class EventManager {

	private static ConcurrentMap<EventKey, List<EventCallback>> callbackMap = new ConcurrentHashMap<>();
	private static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private static Logger log = LoggerFactory.getLogger(EventManager.class);

	@SuppressWarnings("unchecked")
	public static void register(EventKey key, EventCallback callback) {
		if (callbackMap.containsKey(key))
			callbackMap.get(key).add(callback);
		else {
			lock.readLock().lock();
			if (callbackMap.containsKey(key)) {
				callbackMap.get(key).add(callback);
				lock.readLock().unlock();
				return;
			}
			lock.readLock().unlock();
			lock.writeLock().lock();
			callbackMap.put(key, new CopyOnWriteArrayList<>(Collections.singletonList(callback)));
			lock.writeLock().unlock();
		}

		log.debug("event {} registed.", key);
	}

	public static void unregister(EventKey key, EventCallback callback) {
		if (callback == null)
			callbackMap.remove(key);
		else if (callbackMap.containsKey(key))
			callbackMap.get(key).remove(callback);
		log.debug("event {} unregisted.", key);
	}

	@SuppressWarnings("unchecked")
	public static void notify(EventKey key, MqttMessage msg) {
		if (!callbackMap.containsKey(key)) {
			return;
		}

		List<EventCallback> callbackList = new ArrayList<>(callbackMap.get(key));
		for (EventCallback callback : callbackList) {
			callback.callback(msg);
			if (callback instanceof OneTimeEventCallBack)
				unregister(key, callback);
		}

		log.debug("event {} notified", key);
	}

}
